package com.lixixi.tingshu.album.listener;

import com.alibaba.fastjson.JSONObject;
import com.lixixi.tingshu.album.service.AlbumStatService;
import com.lixixi.tingshu.common.constant.KafkaConstant;
import com.lixixi.tingshu.common.constant.RedisConstant;
import com.lixixi.tingshu.vo.album.AlbumStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class AlbumStatUpdateListener {
	@Autowired
	RedisTemplate redisTemplate;

	@Autowired
	AlbumStatService albumStatService;

	@KafkaListener(id = "updateAlbumStat", topics = KafkaConstant.QUEUE_ALBUM_STAT_UPDATE)
	void albumStatListener(String albumStatMqVoJson) {
		AlbumStatMqVo albumStatMqVo = JSONObject.parseObject(albumStatMqVoJson, AlbumStatMqVo.class);
		//使用redis实现分布式锁,标记该消息是否已消费
		//维持10分钟
		Boolean absent = redisTemplate.opsForValue()
				.setIfAbsent(albumStatMqVo.getBusinessNo()
						, 1
						, RedisConstant.CACHE_TEMPORARY_TIMEOUT
						, TimeUnit.SECONDS);
		if (Boolean.FALSE.equals(absent))
		{
			//标识该消息已被消费,不做处理
			return;
		}
		try
		{
			Assert.notNull(albumStatMqVo, "消息传递失败");
			log.info("消息编号:{}-专辑:{}-统计类型:{}-更新:{}"
					, albumStatMqVo.getBusinessNo()
					, albumStatMqVo.getAlbumId()
					, albumStatMqVo.getStatType()
					, albumStatMqVo.getCount());
			albumStatService.updateAlbumStat(albumStatMqVo.getStatType()
					, albumStatMqVo.getCount()
					, albumStatMqVo.getAlbumId());
		} catch (IllegalArgumentException exception)
		{
			//todo:对该消息进行特殊处理
			log.error("消息消费失败:{}", albumStatMqVoJson);
		}
	}
}
